package com.cloud.core.graphx.demo

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD

object Graph_learning_01 {

  var sc: SparkContext = null

  def init(): Unit = {
    val sparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("PageRank_01")

    sc = new SparkContext(sparkConf)
  }

  def main(args: Array[String]): Unit = {

    init()

    val users: RDD[(VertexId, (String, String))] =
      sc.parallelize(Seq((3L, ("rxin", "student")),
        (7L, ("jgonzal", "postdoc")),
        (5L, ("franklin", "prof")),
        (2L, ("istoica", "prof"))))
    // Create an RDD for edges
    val relationships: RDD[Edge[String]] =
      sc.parallelize(Seq(Edge(3L, 7L, "collab"),
        Edge(5L, 3L, "advisor"),
        Edge(5L, 0L, "student"),
        Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))
    // Define a default user in case there are relationship with missing user
    val defaultUser = ("John Doe", "Missing")
    // Build the initial Graph
    val graph = Graph(users, relationships, defaultUser)

    println(graph.vertices.count())

    val facts: RDD[String] =
      graph.triplets.map(triplet =>
        triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
    facts.collect.foreach(println(_))


    val inputGraph =
      graph.outerJoinVertices(graph.outDegrees)((vid, _, degOpt) => degOpt.getOrElse(0))

    inputGraph.vertices.map(v => v._1 + "," + v._2)
      .collect().foreach(println(_))
  }
}
